package com.ndp.fb.mq.consumer.retargeting;

import com.alibaba.dubbo.config.annotation.Reference;
import com.alibaba.fastjson.JSON;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.ndp.ec.core.FeedService;
import com.ndp.fb.cache.redis.serializer.KryoRedisSerializer;
import com.ndp.fb.config.ConfigCentre;
import com.ndp.fb.constant.CommonConstants;
import com.ndp.fb.constant.ConfigConst;
import com.ndp.fb.model.business.bo.consumer.FeedUpdateTemp;
import com.ndp.fb.rdb.model.FeedWe;
import com.ndp.fb.redis.api.RedisClient;
import com.ndp.fb.util.DateFormatUtil;
import com.ndp.fb.util.ThreadPool;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * Created by Evan on 2016/4/21.
 */
@Component
public class FeedCreatingConsumerImpl implements InitializingBean, DisposableBean {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Reference
    FeedService feedService;

    private final KryoRedisSerializer byteSerializer = new KryoRedisSerializer();

    private DefaultMQPushConsumer consumer;

    private String TOP_NAME_PREFIX = ConfigCentre.getString(ConfigConst.MQ_SERVICES_TOPIC_PREFIX);
    private String CONSUMER_GROUP = ConfigConst.MQ_SERVICES_CONSUMER_GROUP;

    private String ROCKETMQ_NAMESRV_DOMAIN = ConfigCentre.getString(ConfigConst.MQ_SERVICES_ROCKETMQ_NAMESRV_DOMAIN);
    private String MQ_SERVICES_ROCKET_MQCLIENT_PASSWORD = ConfigCentre.getString(ConfigConst.MQ_SERVICES_ROCKET_MQCLIENT_PASSWORD);
    private String MQ_SERVICES_LOG_HOME = ConfigCentre.getString(ConfigConst.MQ_SERVICES_LOG_HOME);

    @Autowired
    private RedisClient redisClient;


    @Override
    public void afterPropertiesSet() throws Exception {
        try {
            String topicName;
            if (StringUtils.isNotBlank(TOP_NAME_PREFIX) && TOP_NAME_PREFIX.endsWith("_")) {
                topicName = TOP_NAME_PREFIX + ConfigConst.MQ_SERVICES_FEED_CREATE_TOPIC;
            } else {
                topicName = TOP_NAME_PREFIX + "_" + ConfigConst.MQ_SERVICES_FEED_CREATE_TOPIC;
            }

            String consumerGroup = CONSUMER_GROUP + "_" + topicName;

            System.setProperty(ConfigConst.MQ_SERVICES_ENABLE_SSL, ConfigCentre.getString(ConfigConst.MQ_SERVICES_ENABLE_SSL));
            if (!StringUtils.isBlank(ROCKETMQ_NAMESRV_DOMAIN)) {
                System.setProperty(ConfigConst.MQ_SERVICES_ROCKETMQ_NAMESRV_DOMAIN, ROCKETMQ_NAMESRV_DOMAIN);
            }
            if (!StringUtils.isBlank(MQ_SERVICES_ROCKET_MQCLIENT_PASSWORD)) {
                System.setProperty(ConfigConst.MQ_SERVICES_ROCKET_MQCLIENT_PASSWORD, MQ_SERVICES_ROCKET_MQCLIENT_PASSWORD);
            }
            if (!StringUtils.isBlank(MQ_SERVICES_LOG_HOME)) {
                System.setProperty(ConfigConst.MQ_SERVICES_LOG_HOME, MQ_SERVICES_LOG_HOME);
            }

            logger.info("准备构造消费者 className >>>> " + this.getClass().getName());
            logger.info("consumerGroup >>> " + consumerGroup);
            logger.info("rocketmq.namesrv.domain >>> " + System.getProperty(ConfigConst.MQ_SERVICES_ROCKETMQ_NAMESRV_DOMAIN));
            logger.info("enable_ssl >>> " + System.getProperty(ConfigConst.MQ_SERVICES_ENABLE_SSL));
            logger.info("log.home >>> " + System.getProperty(ConfigConst.MQ_SERVICES_LOG_HOME));
            logger.info("topicName >>> " + topicName);

            consumer = new DefaultMQPushConsumer(consumerGroup);
            consumer.subscribe(topicName, "*");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                    final MessageExt msg = msgs.get(0);
                    final String tags = msg.getTags();
                    final String keys = msg.getKeys();
                    final String redisKey = "mq_key_" + keys;
                    long startTime = System.currentTimeMillis();
                    logger.info("接受到消息 tags >>>> " + tags + " messageKeys >>>>>>  " + keys);
                    try {
                        boolean b = redisClient.setNx(redisKey, 1);
                        if (!b) {
                            long end = System.currentTimeMillis();
                            logger.info("重现重复消费>>>>>>  " + keys + "  >>> 消息处理耗时 >>> " + (end - startTime));
                        } else {
                            redisClient.expire(redisKey, 2, TimeUnit.HOURS.name());
                            ThreadPool.execute(() -> {
                                Object messageBody = byteSerializer.deserialize(msg.getBody());
                                if ((ConfigConst.MQ_SERVICES_FEED_CREATE_TOPIC + ConfigConst.ROCKETMQ_TOPIC_TAG_SPLIT + ConfigConst.MQ_SERVICES_FEED_MERGE_TAG).equals(tags)
                                        && messageBody instanceof FeedWe) {
                                    FeedWe feedWe = (FeedWe) messageBody;
                                    feedService.feedUrlsMergeAndXmlCheck(feedWe);
                                } else if ((ConfigConst.MQ_SERVICES_FEED_CREATE_TOPIC + ConfigConst.ROCKETMQ_TOPIC_TAG_SPLIT + ConfigConst.MQ_SERVICES_FEED_UPDATING).equals(tags)) {
                                    Long weFeedId = (Long) messageBody;
                                    feedService.updating(weFeedId);
                                } else if ((ConfigConst.MQ_SERVICES_FEED_CREATE_TOPIC + ConfigConst.ROCKETMQ_TOPIC_TAG_SPLIT + ConfigConst.MQ_SERVICES_FEED_BUILDING).equals(tags)) {
                                    FeedUpdateTemp obj = (FeedUpdateTemp) messageBody;
                                    if(obj.getCreateAd())
                                        feedService.buildMaterialAndUpdateFbFeedUrlCb(obj);
                                    else
                                        feedService.buildFeedUrlCb(obj);
                                }
                                long end = System.currentTimeMillis();
                                logger.info("消息处理完成messageKeys>>>>>>  " + keys + "  >>> 消息处理耗时 >>> " + (end - startTime));
                            });
                        }
                    } catch (Exception e) {
                        logger.error("消费消息异常", e);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });
            consumer.start();
        } catch (Exception e) {
            logger.error("构造消费者异常", e);
        }
    }

    @Override
    public void destroy() throws Exception {
        if (consumer != null) {
            logger.warn("消费者销毁: " + this.getClass().getName() + "destroy");
            consumer.shutdown();
        }
    }

}
